package cn.mn.ac.rocketmq.producer;

import cn.mn.ac.rocketmq.resolver.PropertyResolver;
import cn.mn.ac.rocketmq.props.RocketProperties;
import cn.mn.ac.rocketmq.annotation.OnsConfiguration;
import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.shade.com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.util.Properties;

/**
 * @author: zhuCan
 * @date: 2019/11/6 18:46
 * @description: 通用的生产者基类
 */
public abstract class DefaultProducerProxy {

    @Autowired
    protected RocketProperties configuration;

    @Autowired
    private PropertyResolver propertyResolver;

    @Autowired
    private SendCallback sendCallback;

    protected Producer producer;

    protected Logger logger = LoggerFactory.getLogger(this.getClass());

    /**
     * 初始化 producer
     */
    @PostConstruct
    public void setup() {
        if (onsConfiguration() == null || !onsConfiguration().enable()) {
            logger.warn("配置开关已关闭!");
            return;
        }
        Properties properties = configuration.rocketProperties();
        properties.put(PropertyKeyConst.GROUP_ID, propertyResolver.springElResolver(configuration.getGroup()));
        producer = ONSFactory.createProducer(properties);
        producer.start();
        logger.info("启动 producer -> group:{}", properties.get(PropertyKeyConst.GROUP_ID));
    }

    // 以下为同步发送

    /**
     * 同步发送
     *
     * @param msg   消息体
     * @param topic topic
     * @param tag   tag
     * @param <T>   消息内容泛型
     * @return 发送结果
     */
    public <T> SendResult sendMsg(T msg, String topic, String tag) {
        printLog(msg);
        return producer.send(new Message(topic, tag, toByte(msg)));
    }

    /**
     * 同步发送
     *
     * @param msg   消息体
     * @param topic topic
     * @param tag   tag
     * @param key   业务主键
     * @param <T>   消息内容泛型
     * @return 发送结果
     */
    public <T> SendResult sendMsg(T msg, String topic, String tag, String key) {
        printLog(msg);
        return producer.send(new Message(topic, tag, key, toByte(msg)));
    }

    /**
     * 同步发送
     *
     * @param message 消息体
     * @return 发送结果
     */
    public SendResult sendMsg(Message message) {
        return producer.send(message);
    }

    // 以下为异步发送

    /**
     * 异步发送消息
     *
     * @param msg   消息体
     * @param topic topic
     * @param tag   tag
     * @param <T>   消息内容泛型
     */
    public <T> void sendMsgAsync(T msg, String topic, String tag) {
        printLog(msg);
        producer.sendAsync(new Message(topic, tag, toByte(msg)), sendCallback);
    }

    /**
     * 异步发送消息
     *
     * @param msg   消息体
     * @param topic topic
     * @param tag   tag
     * @param key   业务主键
     * @param <T>   消息内容泛型
     */
    public <T> void sendMsgAsync(T msg, String topic, String tag, String key) {
        printLog(msg);
        producer.sendAsync(new Message(topic, tag, key, toByte(msg)), sendCallback);
    }

    /**
     * 异步发送
     *
     * @param message 消息体
     */
    public void sendMsgAsync(Message message) {
        this.sendAsync(message);
    }

    // 以下为延时消息 单位: s

    /**
     * 异步发送延时消息
     *
     * @param msg     消息内容
     * @param topic   topic
     * @param tag     tag
     * @param timeout 延时时间
     * @param <T>     消息内容泛型
     */
    public <T> void sendDelayMsgAsync(T msg, String topic, String tag, long timeout) {
        printLog(msg);
        Message message = new Message(topic, tag, toByte(msg));
        message.setStartDeliverTime(System.currentTimeMillis() + timeout * 1000);
        this.sendAsync(message);
    }

    /**
     * 异步发送延时消息
     *
     * @param msg     消息体
     * @param topic   topic
     * @param tag     tag
     * @param key     业务主键
     * @param timeout 延时时间
     * @param <T>     消息内容泛型
     */
    public <T> void sendDelayMsgAsync(T msg, String topic, String tag, String key, long timeout) {
        printLog(msg);
        Message message = new Message(topic, tag, key, toByte(msg));
        message.setStartDeliverTime(System.currentTimeMillis() + timeout * 1000);
        this.sendAsync(message);
    }

    /**
     * 异步发送
     *
     * @param message 消息体
     */
    protected void sendAsync(Message message) {
        producer.sendAsync(message, sendCallback);
    }

    /**
     * json序列化成字节码
     *
     * @param o 消息内容
     * @return 字节数组
     */
    protected byte[] toByte(Object o) {
        return JSON.toJSONString(o).getBytes();
    }

    /**
     * 输出日志
     *
     * @param msg 消息内容
     * @param <T> 消息内容泛型
     */
    private <T> void printLog(T msg) {
        logger.info("{} 生产者发送消息:{} ", getClass().getSimpleName(), msg);
    }

    /**
     * 获取 producer 上面注册的ons配置注解
     *
     * @return 类上注解
     */
    protected OnsConfiguration onsConfiguration() {
        return this.getClass().getAnnotation(OnsConfiguration.class);
    }

}

